package com.googlecode.mailnet.simulation;

import com.googlecode.mailnet.netmodel.Link;
import com.googlecode.mailnet.info.ILinkInfo;
import com.googlecode.mailnet.info.MessageFailureType;
import com.googlecode.mailnet.info.IMessageListener;
import com.googlecode.mailnet.math.RandomUtil;

import java.util.ArrayDeque;

/**
 * @author Iskander Akishev
 */
class LinkState implements ILinkInfo {
	private final SimulationState simulationState;
	final Link link;
	final NodeState sourceState;
	final NodeState destinationState;

	final GradatedMessageQueues queuesForSending;
	final ArrayDeque<Message> transmittingMessages = new ArrayDeque<Message>();
	private Message partiallySentMessage = null;
	private int partiallySentMessagePackagesRemaining = 0;

	private double usageValue;
	/**
	 * Moment of time when this link will be restored; if it is less than or equal to
	 * {@link SimulationState#currentStepNumber} then the link is not broken now.
	 */
	int breakdownEndTime;

    LinkState(SimulationState simulationState, Link link) {
		this.simulationState = simulationState;
		this.link = link;
		this.sourceState = simulationState.nodeStates.get(link.getSource());
		this.destinationState = simulationState.nodeStates.get(link.getDestination());
		sourceState.outgoingLinks.add(this);
		destinationState.incomingLinks.add(this);
		queuesForSending = new GradatedMessageQueues(simulationState.networkConfiguration.getQueueingConfiguration());
	}

	void simulateProbableFailure() {
		if (breakdownEndTime <= simulationState.currentStepNumber) {
            if (RandomUtil.nextBoolean(link.getFailureProbability())) {
				breakdownEndTime = (simulationState.currentStepNumber +
					RandomUtil.nextInteger(link.getCapacityDistribution()));
				
				if (partiallySentMessage != null) {
					partiallySentMessage.drop(MessageFailureType.LINK_FAILURE, this);
					partiallySentMessage = null;
					partiallySentMessagePackagesRemaining = 0;
				}
				while (!transmittingMessages.isEmpty()) {
					transmittingMessages.poll().drop(MessageFailureType.LINK_FAILURE, this);
				}
			}
		}
	}

	public boolean isBroken() {
		return (breakdownEndTime > simulationState.currentStepNumber) ||
			sourceState.isBroken() || destinationState.isBroken();
	}
	
	void deliverMessages() {
		if (isBroken()) {
			assert transmittingMessages.isEmpty();
			return;
		}
		
		while (true) {
			Message message = transmittingMessages.peek();
			if (message == null || message.isStillTransferring()) {
				break;
			}
			transmittingMessages.poll();
			if (destinationState.isBroken() ||
				message.simulateProbableCorruption(link.getPackageCorruptionProbability()))
			{
				message.drop(destinationState.isBroken() ? MessageFailureType.NODE_FAILURE : MessageFailureType.CORRUPTED, destinationState);
			} else {
				message.arriveAt(destinationState);
				for (IMessageListener listener : simulationState.messageListeners) {
					listener.onMessageTransmitted(message, this);
				}
			}
		}
	}

	void consumeMessagesFromQueues() {
		if (isBroken()) {
			return;
		}

		int remainingCapacity = RandomUtil.nextInteger(link.getCapacityDistribution());
        int capacity = remainingCapacity;

		if (partiallySentMessage != null) {
			int part = Math.min(remainingCapacity, partiallySentMessagePackagesRemaining);
			remainingCapacity -= part;
			partiallySentMessagePackagesRemaining -= part;
			if (partiallySentMessagePackagesRemaining == 0) {
				partiallySentMessage.sendThrough(this);
				partiallySentMessage = null;
			}
		}
		while (remainingCapacity > 0) {
			Message message = queuesForSending.pollMessage();
			if (message == null) {
				break;
			}
			if (remainingCapacity >= message.size) {
				message.sendThrough(this);
				remainingCapacity -= message.size;
			} else {
				partiallySentMessage = message;
				partiallySentMessagePackagesRemaining = message.size - remainingCapacity;
                remainingCapacity = 0;
			}
		}
        usageValue = (capacity - remainingCapacity) / (double) capacity;
	}

	// ========== ILinkInfo implementation ==========

    void clearStatistics() {
        queuesForSending.clearStatistics();
        usageValue = 0;
    }

	public int getAttemptedToTransferPerStep() {
        return queuesForSending.bytesAttemptedToSend;
	}

	public Link getLink() {
		return link;
	}

	public double getUsageValue() {
        return usageValue;
	}
}
